/**
 * NSQ Client Module for fibjs
 */
"use strict";
var radon = require('radon');
var util = radon.util;
var nsq = require('nsqjs');

// NSQ消息订阅模块
var Suber = function(option, topic, channel) {
	var self = this;
	/**
	 * readme: https://github.com/dudleycarr/nsqjs/tree/master
	 * @type     {Object}
	 * @property {Array|String} [lookupdHTTPAddresses]  URIs of the nsqlookupd instances.
	 */
	self._option = option;
	self._reader = true;

	self._message_cache = [];
	self._callbacks = [];

}
Suber.prototype.connect = function() {
	var self = this;
	if (!self._ready) {
		self._reader = new nsq.Reader(topic, channel, option);
		self._ready.connect();
		self._reader.on('message', self.onMessage.bind(self));
		self._reader.on('error', self.onError.bind(self));
	}
	return self;
}
Suber.prototype.onMessage = function(msg) {
	var cb = this._callbacks.shift();
	if (cb) {
		cb[0](msg);
	} else {
		this._message_cache.push(msg);
	}
}
Suber.prototype.onError = function(err) {
	var cb;
	while (cb = this._callbacks.shift())
	{
		cb[1](err);
	}
}
Suber.prototype.receive = function(message) {
	var self = this;
	return util.promise(function(done, fail) {
		var msg = self._message_cache.shift()
		if (msg) {
			done(msg);
		} else {
			self._callbacks.push(arguments);
		}
	});
}

// NSQ消息发布者模块
var Puber = function(option) {
	var self = this;
	self._host = option.host || 'localhost';
	self._port = option.port || 4150;
	self._writer = null;
	self._ready = false;
}
Puber.prototype.connect = function() {
	var self = this;
	return util.promise(function(done, fail) {
		if (self._writer) {
			self._writer.removeAllListeners();
			self._writer.close();
		}

		var nsq = self._writer = new nsq.Writer(self._host, self._port);
		nsq.once('ready', function () {
			self._ready = true;
			fail = null;
			done(true);
		});
		nsq.once('closed', function () {
			self._ready = false;
		});
		nsq.once('error', function (err) {
			if (fail)
			{
				fail(err);
			}
			self._writer = null;
			self._ready = false;
		});
		nsq.connect();
	});
}
Puber.prototype.close = function() {
	var self = this;
	self._writer.close();
	self._ready = false;
	return util.promise(function(done, fail) {
		self._writer.on('closed', done);
	});
}
Puber.prototype.pub = util.generator(function *(topic, content) {
	var self = this;
	if (!self._ready) {
		yield self.content();
	}
	return yield util.promise(function(done, fail) {
		self._writer.publish(topic, content, function(err) {
			if (err) {
				fail(radon.error(10000207, err.message));
			} else {
				done(true);
			}
		});
	});
});

exports.nsq = nsq;
exports.puber = Puber;
exports.suber = Suber;